/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.meego.common.gearman;

import com.meego.common.config.Config;
import com.meego.common.log.GA;
import com.meego.common.util.JSonUtils;
import com.meego.common.util.StringUtils;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicLong;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
import org.gearman.client.GearmanClient;
import org.gearman.client.GearmanClientImpl;
import org.gearman.client.GearmanJob;
import org.gearman.client.GearmanJobImpl;
import org.gearman.client.GearmanJobResult;
import org.gearman.common.GearmanJobServerConnection;
import org.gearman.common.GearmanNIOJobServerConnection;

/**
 *
 * @author TueHM
 */
public class GearmanClientManager {

    private static Map<String, GearmanClientManager> instances = new NonBlockingHashMap();
    private final AtomicLong totalQueueErrors = new AtomicLong();
    private ArrayBlockingQueue<GearmanClient> queue;
    private Integer maxQueue;
    private Integer minQueue;
    private String host;
    private Integer port;
    private String function;

    public static GearmanClientManager getInstance(String name) {
        GearmanClientManager instance = (GearmanClientManager) instances.get(name);
        if (instance == null) {
            synchronized (GearmanClientManager.class) {
                if (instance == null) {
                    instance = new GearmanClientManager(name);
                    instances.put(name, instance);
                }
            }
        }
        return instance;
    }

    public GearmanClientManager(String name) {
        this.host = Config.getStrConfig(name + "host", "");
        this.port = Config.getIntConfig(name + "port", 4730);

        this.maxQueue = Config.getIntConfig(name + "maxqueue", 1024);
        this.minQueue = Config.getIntConfig(name + "minqueue", 256);

        this.function = Config.getStrConfig(name + "function", "");

        this.queue = new ArrayBlockingQueue(this.maxQueue.intValue());
    }

    private GearmanClient borrowClient() {
        GearmanClient client = null;
        if (this.queue.size() > 0) {
            synchronized (GearmanClientManager.class) {
                try {
                    client = (GearmanClient) this.queue.take();
                } catch (InterruptedException ex) {
                }
            }
        }
        if (client == null) {
            client = new GearmanClientImpl();
            GearmanJobServerConnection connection = new GearmanNIOJobServerConnection(this.host, this.port.intValue());
            client.addJobServer(connection);
        }

        return client;
    }

    private void returnClient(GearmanClient client) {
        try {
            if ((client != null) && (this.queue.size() <= this.maxQueue.intValue())) {
                this.queue.put(client);
            } else {
                destroyClient(client);
            }
        } catch (InterruptedException e) {
            GA.service.error("Exception in put", e);
        }
    }

    private void destroyClient(GearmanClient client) {
        if (client != null) {
            client.shutdown();
            client = null;
        }
    }

    public boolean push(JobEnt jobIn) {
        String jsonData = JSonUtils.Serialize(jobIn);

        byte[] data = StringUtils.encodeString(jsonData);

        return push(data);
    }

    private boolean push(byte[] data) {
        boolean result;
        GearmanClient client = borrowClient();
        try {
            GearmanJob job = GearmanJobImpl.createBackgroundJob(this.function, data, "");
            client.submit(job);
            GearmanJobResult res = (GearmanJobResult) job.get();
            result = res.jobSucceeded();
            returnClient(client);
        } catch (Exception ex) {
            this.totalQueueErrors.incrementAndGet();
            GA.service.error(ex);
            result = false;
            destroyClient(client);
        }
        return result;
    }

    public Long getTotalErrors() {
        return Long.valueOf(this.totalQueueErrors.get());
    }
}
